//operator to prcess the distinct key word in sql

package qp.operators;

import qp.utils.*;

import java.io.*;
import java.util.*;
import java.lang.*;


//to implement distinct
//first get all batc from base, write them into a local file
// then call external sort to get a sorted tuple file
// 		sort according to the key attribute in tuple	
// get tuple in next(), compare current tuple with previous one to void abanduant
public class Distinct extends Operator{

/*	//the vector contains key attritube in tuple
	//the batch will sorted arroding to these key 
	Vector keyAttrList = null; 
*/	
	// get tuple in next(), compare current tuple with previous one to void abanduant
	Tuple previousTuple = null;
	Tuple currentTuple = null;
	
	Operator base; //the base of Orderby should be project operator
//	Vector attrSet; //the attribute list followin orderby keyward in SQL

	// number of tuples per outbatch
	int batchsize; 
	
	//input buffer and output buffer required during exexution 
	Batch inbatch;
	Batch outbatch;

	int start;       // Cursor position in the input buffer
	boolean eos;  // Indiacate whether end of stream is reached or not
	
	//get the unique file name to store batch in open()
	static int filenum = 0; 
	
	// The file name where the sorted batch object is stored
//	String rfname;   
	// The ObjectInputStream object, just call this object to readObject in next()
	ObjectInputStream sortedInput;

	//flag used in next() to initialise the previousTuple
	boolean realFirstRun = true;
	
	public Distinct(Operator base/*, Vector as*/, int type){
		super(type);
		this.base = base;
//		this.attrSet=as;
		}

/*
	//constructor used in Distinct::clone()
	public Distinct(Operator base, int type, Vector as){
		super(type);
		this.base = base;
		this.keyAttrList=as;
		}
*/		

	//Debug::PPrint(oderby)  need the supprot of 
	//oderby.getBase()
	public Operator getBase()
	{
		return base;
	}

	//RandomOptimizer::makeExecPlan()  
	//the support of Orderby::getBase()  Orderby::setBase()
	public void setBase(Operator base)
	{
		this.base = base;
	}

	//create a unique file name
	private String getUniqueName()
		{
			filenum++; 
			String tempName;
			tempName = "Distinct_" +  String.valueOf(filenum) + ".data";

			return tempName;
		}

/*	//return a vector contain the key attribute from given vector
	private Vector getKeyAttr(Vector attrList)
	{	
		//debug info
		System.out.printf("-------------get the key attribute list-------------\n");
		for(int tempi=0; tempi<attrList.size(); tempi++)
		{
			Debug.PPrint((Attribute)attrList.elementAt(tempi));
		}

		Vector keyAttr = new Vector(attrList.capacity());
							
		for(int i = 0; i < attrList.size(); i++)
		{
			Attribute tempAtt = (Attribute)attrList.elementAt(i);
			
			if(tempAtt.isPrimaryKey() || tempAtt.isForeignKey())
				keyAttr.add(tempAtt);
		}
		//debug info
		System.out.printf("\n==========>\n");
		for(int tempi=0; tempi<keyAttr.size(); tempi++)
		{
			Debug.PPrint((Attribute)keyAttr.elementAt(tempi));
		}
				
		return keyAttr;
	}
*/

	//compare two tuples where is the sam
	//concequently we just support the compare of interger, string
	private boolean isTheSame(Tuple tupleA, Tuple tupleB)//, Vector attrList, Schema sortSchema)
	{
		//compare tuples by the i th attribue in attributelist, attrList.elementAt(i)
		for(int i = 0; i< tupleA.size(); i++)
//		for(int i = 0; i< 1 ; i++)
		{
			Object attribueA = tupleA.dataAt(i);
			Object attribueB = tupleB.dataAt(i);
	
			try{ //try to compare the attribute, while regarding it as integer
				int valueA = ((Integer)attribueA).intValue();			
				int valueB = ((Integer)attribueB).intValue();

				if(valueA!=valueB)
					return false;
			}catch( java.lang.ClassCastException e) //once attribute is string, it will throw this exception
				{
					int tempValue = ((String)attribueA).compareTo((String)attribueB);
					if (tempValue!=0)
						return false;					
				}
		}

		//if they are the same 
		return true;	
	}
		
	public boolean open(){
		eos=false;     // Since the stream is just opened
		start = 0;   // set the cursor to starting position in input buffer
		
		//get the tuples number per outbatch
		int tuplesize = schema.getTupleSize();
		batchsize=Batch.getPageSize()/tuplesize;    /////check when the operator::setscheme will be called for selection, do necessary adjust for oderby 

		//open the base.open()
		if(!base.open())
	    		return false;
		
		//create a unique file name
		String rawBatchFile = this.getUniqueName();
		
		try
		{
			//open a file to stroe the batch
			ObjectOutputStream out = new ObjectOutputStream(new FileOutputStream(rawBatchFile));
			
			//get all the batch from base operator, 
			Batch temp_inbatch;
			while((temp_inbatch = (Batch) base.next()) != null)
			{
				out.writeObject(temp_inbatch);
			}

			//close output file
			out.close();
		}catch (IOException io)
		{
				System.out.println("Oderby:writing the temporary file error");
				return false;
		}

		//get a vector contains key attribue
		Vector allAtribute = this.schema.getAttList();
//		keyAttrList = getKeyAttr(allAtribute);

		//to find the distinct of tuple
		//first external sort for the file accroding to the full list of key attribute
		ExternalSort tempSort = new ExternalSort(rawBatchFile, allAtribute, this.schema);	

		String sortedFile = null;
		try{
			//return the String disply the sorted file,  input file ---> sorted file		
			sortedFile = tempSort.sort();   //, 10, 1);
			//sortedFile = tempSort.sort_debug();
		}catch(IOException io)
		{
			System.out.println("externalsort@orderby throw IOException");
		}catch(ClassNotFoundException classio)
		{
			System.out.println("externalsort@orderby throw ClassNotFoundException");
		}

		//means no tuples meed the condition of query
		if(sortedFile == null)
		{
			//let the next return null, by set this flag as true
			eos = true;
		}
		else //open the input stream
		{
			//open the input stream, it will used to get the batch object in next()
			try{
				sortedInput = new ObjectInputStream(new FileInputStream(sortedFile));
			}catch(IOException io)
			{
			    System.err.println("Objectby::open error in reading the file, sortedFile");
			    System.exit(1);
			}
		}
		
		//close the base.close(), because we have read all the data from base oeprator
		if(!base.close())
	    		return false;	
		else
			return true;		
	
		}

/*
	
	//if there is no more tuple can be found in given file, justreturn null
	//else reture nthe tuple associated with highest priority in given file 
	//		delete the tuple in the file
	//note : file indicated by BatchFIle stored the batch object, batch file store the tuples
	private Tuple GetFirstTuple(String BatchFile, Vector attributeList)
	{
		//in put stream
		try{
			ObjectInputStream in;
			in = new ObjectInputStream(new FileInputStream(BatchFile));
		}catch(IOException io){
			System.err.println("Oderby::GetFirstTuple error in reading the file");
			System.exit(1);
			}
		
		//if no tuple can be found return null
		
		//scan the all batch object in Batch file to find the max tuple 
		//mean while keep the batch, associated with max tuple in menmory

		//delete the max tuple in the batch object and write the batch object back to the file

		//returen the max tuple
	}
	
	//return the bool which tuple is associated with higher priority according to the orderby attribute list
	private boolean CompareTuple(Tuple a, Tuple b, Vector OrderbyList)
	{
		return true;
	}


	public Next(){
		outbatch = new Batch(batchsize);

		//eos is true, means there is no more tuples can be get from inbatch
		if(eos)
		{
			close();
			return null;
		}

		int i =0; //the index used in loop
		
		//loop until outbatch is full
		while(!outbatch.isFull())
		{
			//local file ---> max tuple, according to attribute in orderby list
			Tuple temp_tuple = this.GetFirstTuple(rfname, attrSet);

			//check whether meed the end of input stream
			if (temp_tuple == null) 
			{
				eos=true;  //flag the event, meet the end of input stream
				return outbatch;
			}
			else 
			{
				//max tuple --> outbatch
				outbatch.add(temp_tuple); 
			}

		}

		//return outbatch
		return outbatch;
		}
*/		
	public Batch next(){
		outbatch = new Batch(batchsize);

		//eos is true, means there is no more tuples can be get from inbatch
		if(eos)
		{
			close();
			return null;
		}

		int i =0; //the index used in loop
		
		//loop until outbatch is full
		while(!outbatch.isFull())
		{
			//mean need to read a new batch from local file
			if(start==0)
			{
				//read inbatch from  local file produce in open
				//inbatch = (Batch) base.next();  //debug
				try
				{
//					inbatch.clear();//clear the original input buffer
					inbatch = (Batch)sortedInput.readObject(); //read one page into input buffer from file
				} //mean meet the end of input stream
				catch (EOFException e) 
				{
					System.out.println("Orderby meet the end of file");
					eos = true; 	
					return outbatch;
				}
				catch(IOException io){
		    			System.out.println("Orderby:IOException error");
		    			System.exit(1);
				}
				catch(ClassNotFoundException c)
				{
					    System.out.println("Orderby:Some error in deserialization ");
					    System.exit(1);
				}
				
			}

			//initialize the previousTuple
			if(realFirstRun)
			{
				realFirstRun=false;
				previousTuple = inbatch.elementAt(0);
			}
			//loop until outbatch is full/ inbatch is finished
			for(i=start; (i<inbatch.size()) && (! outbatch.isFull()); i++)
			{
				 currentTuple = inbatch.elementAt(i);
				 if (isTheSame(currentTuple, previousTuple)) //currentTuple == previousTuple
				 {
				 	System.out.printf("\n-----------------DISTINCT, following two tuples are the same----------------\n");
///*
					Debug.PPrint(previousTuple);
				 	System.out.printf("-----------------------\n");
					Debug.PPrint(currentTuple);
					System.out.printf("\n");

//*/					
				 	previousTuple = currentTuple;
					currentTuple = null;
				 }
				 else //currentTuple != previousTuple
				 {
				 	outbatch.add(currentTuple);
					previousTuple = currentTuple;
					currentTuple = null;
				 }

			}

			//matain the cursor
			if(i==inbatch.size())
				start=0;
			else
				start=i;
		}

		//return outbatch
		return outbatch;
		}

	public boolean close()
		{
			return true;
		}

	public Object clone(){
		//copy the base object
		Operator newbase = (Operator) base.clone();

		//copy the relatived Object in current object
/*		Vector newattr = new Vector();
		for (int i=0; i<keyAttrList.size(); i++)
				newattr.add((Attribute) ((Attribute)keyAttrList.elementAt(i)).clone()) ;
		Distinct neworder = new Distinct(newbase,optype,newattr);
*/		
		//initialise a new project operator
		Distinct neworder = new Distinct(newbase,optype);


		//copy the sechme from base to operator
		//newbase -> schme -> newproject
		
		////oderby need all the attribule. 
		//the schema related code should be the same 
		// between RandomInitialPlan::createOrderbyOp , oderby::clone(),  RandomOptimizer::modifySchem(root)
		Schema newSchema = newbase.getSchema();
		neworder.setSchema(newSchema);

		return neworder;
		}
	
}
